package com.luych.toolbox.data.service.job;

import com.luych.toolbox.common.enums.DataTransporterHistoryStatus;
import com.luych.toolbox.common.enums.DataTransporterPlanStatus;
import com.luych.toolbox.data.service.constant.Constant;
import com.luych.toolbox.data.service.dao.DataTransporterHistoryDao;
import com.luych.toolbox.data.service.dao.DataTransporterPlanDao;
import com.luych.toolbox.data.service.entity.DataTransporterHistory;
import com.luych.toolbox.data.service.entity.DataTransporterPlan;
import org.quartz.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.scheduling.quartz.QuartzJobBean;

import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;

public class DataTransporterPlanExecuteJob extends QuartzJobBean {
    private static final Logger LOGGER = LoggerFactory.getLogger(DataTransporterPlanExecuteJob.class);

    private final Scheduler scheduler;

    private final DataTransporterPlanDao dataTransporterPlanDao;
    private final DataTransporterHistoryDao dataTransporterHistoryDao;

    public DataTransporterPlanExecuteJob(Scheduler scheduler, DataTransporterPlanDao dataTransporterPlanDao,
                                         DataTransporterHistoryDao dataTransporterHistoryDao) {
        this.scheduler = scheduler;
        this.dataTransporterPlanDao = dataTransporterPlanDao;
        this.dataTransporterHistoryDao = dataTransporterHistoryDao;
    }

    private JobDetail getJobDetail(String id) {
        Map<String, Object> jobParam = new HashMap<String, Object>() {{
            this.put("id", id);
        }};
        return JobBuilder.newJob(DataTransporterHistoryExecuteJob.class)
                .withIdentity(id, DataTransporterHistoryExecuteJob.class.getName())
                .usingJobData(new JobDataMap(jobParam))
                .build();
    }

    private Trigger getJobTrigger(String id) {
        return TriggerBuilder.newTrigger()
                .withIdentity(id, DataTransporterHistoryExecuteJob.class.getName())
                .withSchedule(SimpleScheduleBuilder.repeatSecondlyForTotalCount(1, 1))
                .build();
    }

    @Override
    protected void executeInternal(JobExecutionContext context) throws JobExecutionException {
        String id = context.getJobDetail().getJobDataMap().getString("id");

        DataTransporterPlan dataTransporterPlan = dataTransporterPlanDao.findById(id).orElseThrow(() ->
                new JobExecutionException(Constant.DATA_TRANSPORTER_PLAN_NOT_FOUND));

        if (DataTransporterPlanStatus.DISABLE.equals(dataTransporterPlan.getStatus())) {
            LOGGER.warn("the data transporter plan[{}] is disabled, but still run in quartz, why ???", dataTransporterPlan.getId());
            return;
        }

        DataTransporterHistory transporterHistory = new DataTransporterHistory();
        transporterHistory.setId(UUID.randomUUID().toString());
        transporterHistory.setCreateUserId(dataTransporterPlan.getCreateUserId());
        transporterHistory.setCreateDate(new Date());
        transporterHistory.setName(dataTransporterPlan.getName());
        transporterHistory.setTables(dataTransporterPlan.getTables());
        transporterHistory.setStatus(DataTransporterHistoryStatus.LOADING);
        transporterHistory.setDataSourceFromId(dataTransporterPlan.getDataSourceFromId());
        transporterHistory.setDataSourceToId(dataTransporterPlan.getDataSourceToId());
        transporterHistory.setDataTransporterPlanId(dataTransporterPlan.getId());
        transporterHistory.setOwnerId(dataTransporterPlan.getOwnerId());
        dataTransporterHistoryDao.save(transporterHistory);

        LOGGER.info("start to schedule data transporter plan [{}]...", dataTransporterPlan.getName());
        try {
            scheduler.scheduleJob(this.getJobDetail(transporterHistory.getId()), this.getJobTrigger(transporterHistory.getId()));
            LOGGER.info("success to schedule data transporter plan [{}]", dataTransporterPlan.getName());
        } catch (SchedulerException e) {
            LOGGER.info("error to schedule data transporter plan [{}]: {}", dataTransporterPlan.getName(), e.getMessage(), e);
            throw new JobExecutionException(e);
        }
    }
}
